package com.taotao.async.task.result;

import com.alibaba.fastjson.JSON;
import com.taotao.async.task.TaskExeContext;
import com.taotao.async.task.recovery.RecoveryMessage;
import com.taotao.async.task.recovery.RecoveryStore;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 *  结果调度
 * </p>
 *
 * @author fanbeibei
 * @version 1.0.0
 * @date 2024/6/27 18:24
 */
@Slf4j
public class TaskResultScheduler {

    private static final int BATCH = 50;

    private RecoveryStore recoveryStore;

    private TaskResultProccessor taskResultProccessor;

    private ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(2);

    public TaskResultScheduler(String path, TaskResultProccessor taskResultProccessor) {
        this.recoveryStore = new RecoveryStore(path);
        this.taskResultProccessor = taskResultProccessor;
    }

    public TaskResultScheduler(RecoveryStore recoveryStore, TaskResultProccessor taskResultProccessor) {
        this.recoveryStore = recoveryStore;
        this.taskResultProccessor = taskResultProccessor;
    }

    public void init() {
        scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    List<RecoveryMessage> list = recoveryStore.queryList(BATCH);
                    if (null == list || list.isEmpty()) {
                        break;
                    }

                    for (RecoveryMessage m : list) {
                        proccess(m);
                    }
                }
            }

        }, 5, 30, TimeUnit.SECONDS);
    }


    private void proccess(RecoveryMessage m) {
        try {
            TaskExeContext ctx = JSON.parseObject(m.getJson(), TaskExeContext.class);
            taskResultProccessor.proccessTaskResult(ctx);
            recoveryStore.delete(m.getKey());
        } catch (Exception e) {
        }
    }

    public void addResult(String taskInstId, TaskExeContext context) {
        try {
            recoveryStore.put(taskInstId, new RecoveryMessage(taskInstId, 0, JSON.toJSONString(context)));
        } catch (Exception e) {
            log.error("",e);
        }
    }


    public void proccess(TaskExeContext context) {
        try {
            taskResultProccessor.proccessTaskResult(context);
        } catch (Exception e) {
            addResult(context.getTaskInstId().toString(),context);
            log.error("",e);
        }
    }

}
